package com.kancy.stream.rabbit.service.impl;

import com.kancy.stream.rabbit.config.DynamicInterfaceBeanResolver;
import com.kancy.stream.rabbit.converter.MessageConverter;
import com.kancy.stream.rabbit.exception.MessageException;
import com.kancy.stream.rabbit.exception.NotFoundChannelKeyMessageException;
import com.kancy.stream.rabbit.exception.SendMessageException;
import com.kancy.stream.rabbit.message.ChannelMessage;
import com.kancy.stream.rabbit.properties.ChannelProperties;
import com.kancy.stream.rabbit.properties.RabbitMessageProperties;
import com.kancy.stream.rabbit.service.MessageChannelService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * MessageChannelServiceImpl
 *
 * @author kancy
 * @date 2020/3/8 12:02
 */
public class MessageChannelServiceImpl extends AbstractRabbitService implements MessageChannelService {

    private final RabbitMessageProperties rabbitMessageProperties;
    private final DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver;

    public MessageChannelServiceImpl(RabbitTemplate rabbitTemplate,
                                     RabbitMessageProperties rabbitMessageProperties,
                                     DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver) {
        super(rabbitTemplate);
        this.rabbitMessageProperties = rabbitMessageProperties;
        this.dynamicInterfaceBeanResolver = dynamicInterfaceBeanResolver;
    }

    /**
     * 发送消息
     *
     * @param message
     */
    @Override
    public void sendMessage(ChannelMessage message) {
        String channelKey = message.getChannelKey();
        Assert.hasText(channelKey, "ChannelKey is empty");

        if (rabbitMessageProperties.getChannel().containsKey(channelKey)){
            ChannelProperties channelProperties = rabbitMessageProperties.getChannel().get(channelKey);
            String messageType = message.getMessageType();

            MessageConverter messageConverter = null;
            if(StringUtils.hasText(messageType)){
                messageConverter = geMessageConverter(channelProperties, messageType);
            }else{
                messageConverter = dynamicInterfaceBeanResolver.getBean(
                        channelProperties.getMessageConverter(), MessageConverter.class);
            }
            MessageProperties properties = new MessageProperties();
            properties.setCorrelationId(message.getId());
            properties.getHeaders().putAll(message.getHeaders());
            Message toMessage = messageConverter.toMessage(message.getPayload(), properties);
            sendMessage(channelProperties, toMessage);
        }else{
            throw new NotFoundChannelKeyMessageException(String.format("Channel message [%s] Not Found", channelKey));
        }
    }

    /**
     * 获取消息转换器
     * @param channelProperties
     * @param messageType
     * @return
     */
    private MessageConverter geMessageConverter(ChannelProperties channelProperties, String messageType) {
        ChannelProperties.ProducerProperties producerProperties = channelProperties.getProducer().get(messageType);
        if(Objects.isNull(producerProperties)){
            return dynamicInterfaceBeanResolver.getBean(channelProperties.getMessageConverter(), MessageConverter.class);
        }
        return dynamicInterfaceBeanResolver.getBean(producerProperties.getMessageConverter(), channelProperties.getMessageConverter(),
                            MessageConverter.class);
    }

    /**
     * 向渠道发送消息
     * @param channelConsumerProperties
     * @param message
     */
    private void sendMessage(ChannelProperties channelConsumerProperties, Message message) {
        RabbitMessageProperties.BindingQueueProperties bindingQueueProperties = rabbitMessageProperties.getBindingQueue().get(channelConsumerProperties.getBindingKey());

        String exchangeName = null;
        String routingKey = null;
        if (Objects.isNull(bindingQueueProperties)){
            RabbitMessageProperties.BingProperties bingProperties = rabbitMessageProperties.getBinding().get(channelConsumerProperties.getBindingKey());
            if (Objects.nonNull(bingProperties)){
                exchangeName = bingProperties.getExchangeName();
                routingKey = bingProperties.getRoutingKey();
            }
        }else{
            exchangeName = bindingQueueProperties.getExchangeName();
            routingKey = bindingQueueProperties.getRoutingKey();
        }

        if (StringUtils.isEmpty(exchangeName) || StringUtils.isEmpty(routingKey)){
            throw new MessageException(String.format(
                    "Binding key [%s] is error, not found exchange and routing key.", channelConsumerProperties.getBindingKey()));
        }

        // 发送消息
        CorrelationData correlationData = new CorrelationData(message.getMessageProperties().getCorrelationId());
        try {
            rabbitTemplate.send(exchangeName, routingKey, message, correlationData);
        } catch (Exception e) {
            throw new SendMessageException(String.format("消息[%s]发送失败", correlationData.getId()), e);
        }
    }
}
